package com.flyfinger.client;

import com.flyfinger.codec.RpcDecoder;
import com.flyfinger.codec.RpcEncoder;
import com.flyfinger.codec.Serializer;
import com.flyfinger.codec.impl.JSONSerializer;
import com.flyfinger.handler.UserClientHandler;
import com.flyfinger.pojo.RpcRequest;
import com.flyfinger.pojo.RpcRequestUtil;
import com.flyfinger.pojo.ServiceLocation;
import com.flyfinger.register.CuratorClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;

import java.io.IOException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 消费者
 */
public class RPCConsumer {

	private final ExecutorService executorService =
			Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

	private final Serializer serializer = JSONSerializer.getInstance();

	private final Map<String, ServiceLocation> serviceLocationRegister = new TreeMap<>();

	private final AtomicInteger index = new AtomicInteger(-1);

	private UserClientHandler userClientHandler;

	CuratorFramework curatorClient;
	CuratorCache curatorCacheClient;

	public void close() {
		closeZK();
		if (userClientHandler != null) {
			userClientHandler.close();
		}
	}

	public RPCConsumer() throws Exception {
		initZKNLocationCache();
	}


	private void initNetty(String host, int port) throws Exception {
		if (userClientHandler != null) {
			userClientHandler.close();
		}
		userClientHandler = new UserClientHandler();
		Bootstrap bootstrap = new Bootstrap();
		bootstrap.group(new NioEventLoopGroup())
				.channel(NioSocketChannel.class)
				.option(ChannelOption.TCP_NODELAY, true)
				.handler(new ChannelInitializer<SocketChannel>() {
					protected void initChannel(SocketChannel socketChannel) throws Exception {
						ChannelPipeline pipeline = socketChannel.pipeline();
						pipeline.addFirst(new RpcDecoder(serializer));
						pipeline.addLast(new RpcEncoder(serializer));
						pipeline.addLast(userClientHandler);
					}
				});
		bootstrap.connect(host, port).sync();
	}

	@SuppressWarnings("unchecked")
	public <T> T createProxy(Class<? extends T> serviceClass) {
		return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
				new Class[]{serviceClass},
				(Object proxy, Method method, Object[] args) -> {
					final RpcRequest rpcRequest = RpcRequestUtil
							.initRpcRequestByMethodName(serviceClass, method.getName(), args);

					final ServiceLocation location = getServiceLocation();
					initNetty(location.getIp(), location.getPort());

					userClientHandler.setRpcRequest(rpcRequest);

					final Future<String> submit = executorService.submit(userClientHandler);

					return submit.get();
				});
	}

	private void initZKNLocationCache() throws Exception {
		curatorClient = initServiceLocationRegister();
		curatorCacheClient = initZKListener(curatorClient);
	}

	private void closeZK() {
		curatorClient.close();
		curatorCacheClient.close();
	}

	private CuratorCache initZKListener(CuratorFramework client) {
		final CuratorCache curatorCache = CuratorCache
				.builder(client, CuratorClient.ZK_CLIENT_PATH)
				.build();
		curatorCache.start();

		final Listenable<CuratorCacheListener> listenable = curatorCache.listenable();

		final CuratorCacheListener listener = CuratorCacheListener.builder()
				.forDeletes(childData -> {
					final ServiceLocation location = catchChildData(childData);
					deleteLocationRegistration(location);
				})
				.forCreates(childData -> {
					final ServiceLocation location = catchChildData(childData);
					addLocationRegistration(location);
				}).build();

		listenable.addListener(listener);
		return curatorCache;
	}

	private CuratorFramework initServiceLocationRegister() throws Exception {
		serviceLocationRegister.clear();
		final CuratorFramework client = CuratorClient.curator();
		client.start();

		final List<String> strings = client.getChildren()
				.watched()
				.forPath(CuratorClient.ZK_CLIENT_PATH);

		for (String path : strings) {
			final byte[] bytes = client.getData()
					.forPath(buildPath(
							CuratorClient.ZK_CLIENT_PATH, "/", path));

			final ServiceLocation location = catchChildData(bytes);

			serviceLocationRegister.put(location.toString(), location);
		}
		return client;
	}

	public void deleteLocationRegistration(ServiceLocation location) {
		System.out.println("++++++++++++++++++++++++++++++++++");
		System.out.println(
				String.format("“下线”服务，服务地址为：[%s]", location));
		System.out.println("++++++++++++++++++++++++++++++++++");
		serviceLocationRegister.remove(location.toString());
	}

	public void addLocationRegistration(ServiceLocation location) {
		System.out.println("++++++++++++++++++++++++++++++++++");
		System.out.println(
				String.format("“上线”服务，服务地址为：[%s]", location));
		System.out.println("++++++++++++++++++++++++++++++++++");
		serviceLocationRegister.put(location.toString(), location);
	}

	private ServiceLocation catchChildData(ChildData childData) {
		final byte[] data = childData.getData();
		return catchChildData(data);
	}

	private ServiceLocation catchChildData(byte[] data) {
		ServiceLocation location;
		try {
			location = JSONSerializer.getInstance().deserialize(ServiceLocation.class, data);
		} catch (IOException e) {
			throw new RuntimeException("解析zk删除节点数据失败");
		}
		if (location == null) {
			throw new RuntimeException("未获取到zk删除的节点数据");
		}
		return location;
	}

	private ServiceLocation getServiceLocation() {
		index.getAndIncrement();
		final Set<String> keySet = serviceLocationRegister.keySet();
		if (index.get() >= keySet.size()) {
			index.set(0);
		}
		int j = 0;
		int i = index.get();
		for (String key : keySet) {
			if (j == i) {
				return serviceLocationRegister.get(key);
			}
			j++;
		}
		throw new RuntimeException("没找到ip和端口");
	}

	private String buildPath(String... param) {
		final StringBuilder builder = new StringBuilder();
		for (String s : param) {
			builder.append(s);
		}
		return builder.length() == 0 ? null : builder.toString();
	}

}
